﻿/**
 * Copyright (c) 2013 Jeremy Wyman
 * Microsoft Public License (Ms-PL)
 * This license governs use of the accompanying software. If you use the software, you accept this license. If you do not accept the license, do not use the software.
 * 1. Definitions
 *    The terms "reproduce", "reproduction", "derivative works", and "distribution" have the same meaning here as under U.S. copyright law.
 *    A "contribution" is the original software, or any additions or changes to the software.
 *    A "contributor" is any person that distributes its contribution under this license.
 *    "Licensed patents" are a contributor's patent claims that read directly on its contribution.
 * 2. Grant of Rights
 *    (A) Copyright Grant - Subject to the terms of this license, including the license conditions and limitations in section 3, each contributor grants you a non-exclusive, worldwide, royalty-free copyright license to reproduce its contribution, prepare derivative works of its contribution, and distribute its contribution or any derivative works that you create.
 *    (B) Patent Grant - Subject to the terms of this license, including the license conditions and limitations in section 3, each contributor grants you a non-exclusive, worldwide, royalty-free license under its licensed patents to make, have made, use, sell, offer for sale, import, and/or otherwise dispose of its contribution in the software or derivative works of the contribution in the software.
 * 3. Conditions and Limitations
 *    (A) No Trademark License - This license does not grant you rights to use any contributors' name, logo, or trademarks.
 *    (B) If you bring a patent claim against any contributor over patents that you claim are infringed by the software, your patent license from such contributor to the software ends automatically.
 *    (C) If you distribute any portion of the software, you must retain all copyright, patent, trademark, and attribution notices that are present in the software.
 *    (D) If you distribute any portion of the software in source code form, you may do so only under this license by including a complete copy of this license with your distribution. If you distribute any portion of the software in compiled or object code form, you may only do so under a license that complies with this license.
 *    (E) The software is licensed "as-is." You bear the risk of using it. The contributors give no express warranties, guarantees or conditions. You may have additional consumer rights under your local laws which this license cannot change. To the extent permitted under your local laws, the contributors exclude the implied warranties of merchantability, fitness for a particular purpose and non-infringement.
 *    
 * More info on: http://tempest.codeplex.com
**/

using System;
using System.Collections.Generic;
using System.Threading;
using System.Text;

using Debug = System.Diagnostics.Debug;
using Stopwatch = System.Diagnostics.Stopwatch;
using Task = System.Threading.Tasks.Task;

namespace Tempest.Threading
{
    /// <summary>
    /// Queue of items to be worked on asynchronously. Add work to the queue and the internal worker threads will work on each item in a FIFO manner. Supports suspend/resume and stop functionality.
    /// </summary>
    /// <typeparam name="TWork">The type of the work items</typeparam>
    [System.Diagnostics.DebuggerDisplay("WorkQueue Count={Count}/{Capacity}, Workers={Workers}")]
    internal class WorkQueue<TWork>
    {
        #region Constants
        private const long GateOpened = 0; // signals that the worker queue is working
        private const long GateBlocked = 1; // signals that the worker queue is paused
        private const long GateClosed = 2; // signals that the worker queue is stopping
        #endregion
        #region Constructors
        /// <summary>
        /// Creates a new WorkQueue
        /// </summary>
        /// <param name="workerMethod">The method the worker threads will use to process each work item</param>
        /// <param name="maxWorkers">The maximum number of workers active at any one time</param>
        /// <param name="maxCapacity">The maximum number of items that can be pending in the queue at any one time</param>
        public WorkQueue(Action<TWork> workerMethod, int maxWorkers, int maxCapacity)
        {
            #region Parameter Validation
            if (workerMethod == null)
                throw new ArgumentNullException();
            if (maxWorkers < 1)
                throw new ArgumentOutOfRangeException();
            if (maxCapacity < 1)
                throw new ArgumentOutOfRangeException();
            #endregion
            _maxCapacity = maxCapacity;
            _maxWorkers = maxWorkers;
            _workerMethod = workerMethod;
            _workQueue = new Queue<TWork>(maxCapacity);
            _workerThreads = new Dictionary<int, Thread>(maxWorkers);
            _workingWorkers = 0;
            _masterThreadId = Thread.CurrentThread.ManagedThreadId;
            // workers should begin immediately
            lock (_workerThreads)
            {
                for (int i = 0; _workerThreads.Count < _maxWorkers; i++)
                {
                    Thread t = new Thread(new ThreadStart(this.Work));
                    t.Name = "WorkQueue.Worker";
                    _workerThreads.Add(t.ManagedThreadId, t);
                    t.Start();
                }
            }
            _workerGate = GateOpened;
            lock (_workQueue)
            {
                Monitor.PulseAll(_workQueue);
            }
        }
        /// <summary>
        /// Creates a new WorkQueue
        /// </summary>
        /// <param name="workerMethod">The method the worker threads will use to process each work item</param>
        /// <param name="maxWorkers">The maximum number of workers active at any one time</param>
        public WorkQueue(Action<TWork> workerMethod, int maxWorkers) : this(workerMethod, maxWorkers, maxWorkers * 16) { }
        /// <summary>
        /// Creates a new WorkQueue
        /// </summary>
        /// <param name="workerMethod">The method the worker threads will use to process each work item</param>
        public WorkQueue(Action<TWork> workerMethod) : this(workerMethod, Environment.ProcessorCount) { }
        #endregion
        #region Members
        /// <summary>
        /// Gets the maximum number of queued work items
        /// </summary>
        public int Capacity
        {
            get { return _maxCapacity; }
        }
        /// <summary>
        /// Gets the number of work items currently in the queue
        /// </summary>
        public int Count { get { lock (_workQueue) { return _workQueue.Count; } } }
        /// <summary>
        /// Sets the method by which exceptions in the worker method are handled. If unset exceptions will cause an application error and stop the worker queue.
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if set from a thread other than the thread it was created on
        /// </exception>
        public Action<Exception, TWork> ExceptionHandler
        {
            set
            {
                this.ValidateCallingThread();
                lock (_workerThreads)
                {
                    _exceptionHandler = value;
                }
            }
        }
        /// <summary>
        /// Gets if the WorkQueue has been stopped or not started
        /// </summary>
        public bool IsStopped { get { return Interlocked.Read(ref _workerGate) == GateClosed; } }
        /// <summary>
        /// Gets if the WorkQueue has been suspended
        /// </summary>
        public bool IsSuspended { get { return Interlocked.Read(ref _workerGate) == GateBlocked; } }
        /// <summary>
        /// Gets if the WorkQueue has been started and not suspended or stopped
        /// </summary>
        public bool IsWorking { get { return Interlocked.Read(ref _workerGate) == GateOpened; } }
        /// <summary>
        /// Gets or sets the number of workers assigned to the WorkQueue
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if set from a thread other than the thread it was created on
        /// </exception>
        public int Workers
        {
            get { lock (_workerThreads) { return _workerThreads.Count; } }
            set
            {
                if (value <= 0)
                    throw new ArgumentOutOfRangeException("There must be atleast one worker");
                this.ValidateCallingThread();
                Interlocked.Exchange(ref _maxWorkers, value);
                lock (_workerThreads)
                {
                    while ((_maxWorkers - _workerThreads.Count) > 0)
                    {
                        Thread t = new Thread(new ThreadStart(this.Work));
                        t.Name = "WorkQueue.Worker";
                        _workerThreads.Add(t.ManagedThreadId, t);
                        t.Start();
                    }
                }
            }
        }
        #region Private Members
        private int _maxCapacity;
        private int _minCapacity;
        private int _maxWorkers;
        private long _workingWorkers;
        private int _masterThreadId;
        [System.Diagnostics.DebuggerBrowsable(System.Diagnostics.DebuggerBrowsableState.Never)]
        private Queue<TWork> _workQueue;
        [System.Diagnostics.DebuggerBrowsable(System.Diagnostics.DebuggerBrowsableState.Never)]
        private Dictionary<int, Thread> _workerThreads;
        private Action<TWork> _workerMethod;
        private Func<int, List<TWork>> _workFetchMethod;
        private Action<Exception, TWork> _exceptionHandler;
        /* Interlocked Gates */
        [System.Diagnostics.DebuggerBrowsable(System.Diagnostics.DebuggerBrowsableState.Never)]
        private long _workerGate = GateClosed;
        [System.Diagnostics.DebuggerBrowsable(System.Diagnostics.DebuggerBrowsableState.Never)]
        private long _fetchGate = GateOpened;
        #endregion
        #endregion
        #region Events
        /// <summary>
        /// 
        /// </summary>
        public event Action<WorkQueue<TWork>> Resumed;
        /// <summary>
        /// 
        /// </summary>
        public event Action<WorkQueue<TWork>> Stopped;
        /// <summary>
        /// 
        /// </summary>
        public event Action<WorkQueue<TWork>> Suspended;
        #endregion
        #region Methods
        /// <summary>
        /// Adds work to the queue if there is sufficient room. Non-blocking
        /// </summary>
        /// <returns>Returns true if work was added to the queue; false if there was insufficient room in the queue</returns>
        public bool AddWork(TWork work)
        {
            lock (_workQueue)
            {
                if (_workQueue.Count >= _maxCapacity)
                    return false;
                // add the work to the queue
                _workQueue.Enqueue(work);
                // signal all blocked threads that the queue has changed
                Monitor.PulseAll(_workQueue);
            }
            return true;
        }
        /// <summary>
        /// Adds work to the queue. Blocks until sufucient room is available in the queue.
        /// </summary>
        /// <param name="millisecondsTimeout">The maximum length of time (in milliseconds) the caller is willing to wait. -1 is inifinity</param>
        /// <returns>Returns true if work was added to the queue; false if the attempt timed out and no work was added</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool AddWork(TWork work, int millisecondsTimeout)
        {
            #region Parameter Validation
            if (millisecondsTimeout < -1)
                throw new ArgumentOutOfRangeException("Invalid timeout value. Valid values are greater than or equal to 0; or -1 (inifinity).");
            #endregion
            this.ValidateCallingThread();
            // create a time, lock the queue, and add work
            Stopwatch timer = new Stopwatch();
            lock (_workQueue)
            {
                // wait for space to be made in the queue
                timer.Start();
                while (_workQueue.Count >= _maxCapacity)
                {
                    // if the queue is stopped or suspended, space won't get made: exit now or deadlock
                    if (Interlocked.Read(ref _workerGate) != GateOpened)
                        return false;
                    // wait for space to be created
                    if (millisecondsTimeout == Timeout.Infinite)
                    {
                        Monitor.Wait(_workQueue, Timeout.Infinite);
                    }
                    else
                    {
                        // use monitor's wait timeout, but also need to track actual time spent as other threads might be signaling and causing wait call to be restarted
                        if (timer.ElapsedMilliseconds >= millisecondsTimeout || !Monitor.Wait(_workQueue, (int)(millisecondsTimeout - timer.ElapsedMilliseconds)))
                            return false; // <-- time elapsed
                    }
                }
                // stop the timer and enqueue the work
                timer.Stop();
                _workQueue.Enqueue(work);
                // signal all blocked threads that the queue has changed
                Monitor.PulseAll(_workQueue);
            }
            return true;
        }
        /// <summary>
        /// Enables the WorkQueue.WorkerMethod to check for Suspend requests. Blocks if the WorkQueue has been suspended. Can only be called by worker threads
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// Throws a InvalidOperationException if called by a non-worker thread
        /// </exception>
        public void Check()
        {
            this.ValidateWorkerThread();
            //
            lock (_workQueue)
            {
                Interlocked.Decrement(ref _workingWorkers);
                while (Interlocked.Read(ref _workerGate) == GateBlocked)
                {
                    Monitor.PulseAll(_workQueue);
                    Monitor.Wait(_workQueue);
                }
                Interlocked.Increment(ref _workingWorkers);
            }
        }
        /// <summary>
        /// Diables the WorkQueue from requesting more work. Non-blocking
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        /// /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed if the Fetching is not enabled
        /// </exception>
        public void DisableFetching()
        {
            this.ValidateCallingThread();
            //
            lock (_workerThreads)
            {
                if (_workFetchMethod == null)
                    throw new InvalidOperationException("Cannot disable work fetching before it has been enabled.");
                _workFetchMethod = null;
                _minCapacity = 0;
            }
        }
        /// <summary>
        /// Dumps all remaining work items, flushing the queue. The queue must be stopped before work can be dumped
        /// </summary>
        /// <returns>Queue of remaining work items</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed when the WorkQueue has been not stopped first
        /// </exception>
        public Queue<TWork> DumpQueue()
        {
            this.ValidateCallingThread();
            //
            lock (_workQueue)
            {
                switch (Interlocked.Read(ref _workerGate))
                {
                    case GateBlocked:
                        throw new InvalidOperationException("Can only dump work from a stopped queue.");
                    case GateClosed:
                        {
                            this.Wait(Timeout.Infinite);
                            Queue<TWork> q = _workQueue;
                            _workQueue = new Queue<TWork>(_maxCapacity);
                            return q;
                        }
                    case GateOpened:
                        throw new InvalidOperationException("Can only dump work from a stopped queue.");
                    default:
                        throw new InvalidOperationException("Invalid WorkQueue state encountered");
                }
            }
        }
        /// <summary>
        /// Enables the WorkQueue to request more work when the current queue gets too low. Non-blocking
        /// </summary>
        /// <param name="workFetchMethod">Method that takes the available space in the queue and returns a list of work items to place into the queue.</param>
        /// <param name="minQueueCount">The lowest number of items in the queue allowed before more work should be requested</param>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public void EnableFetching(Func<int, List<TWork>> workFetchMethod, int minQueueCount)
        {
            #region Parameter Validation
            if (workFetchMethod == null)
                throw new ArgumentNullException("workFetchMethod");
            if (minQueueCount < 0)
                throw new ArgumentOutOfRangeException("minQueueCount");
            #endregion
            this.ValidateCallingThread();
            //
            lock (_workerThreads)
            {
                _workFetchMethod = workFetchMethod;
                _minCapacity = minQueueCount;
            }
        }
        /// <summary>
        /// Completes as much remaining work as possible within the timeout, then stops the queue
        /// </summary>
        /// <param name="millisecondsTimeout"></param>
        /// <returns>Returns true if the queue completed before the timeout limit; returns false if the queue didn't complete before the timeout.</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Flush(int millisecondsTimeout)
        {
            #region Parameter Validation
            if (millisecondsTimeout < -1)
                throw new ArgumentOutOfRangeException("Invalid timeout value. Valid values are greater than or equal to 0; or -1 (inifinity).");
            #endregion
            this.ValidateCallingThread();
            // wait for the queue to complete
            bool completed = this.Wait(millisecondsTimeout);
            // stop all workers, regardless if the queue is complete
            this.Stop();
            return completed;
        }
        /// <summary>
        /// Signals all workers to resume working. Non-blocking
        /// </summary>
        /// <returns>Returns true when all workers are resumed</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Resume()
        {
            this.ValidateCallingThread();
            // unblock the worker gate
            switch (Interlocked.CompareExchange(ref _workerGate, GateOpened, GateBlocked))
            {
                case GateBlocked:
                    {
                        // signal all waiting threads that state has changed
                        lock (_workQueue)
                        {
                            Monitor.PulseAll(_workQueue);
                        }
                    } break;
                case GateClosed:
                    throw new InvalidOperationException("Cannot call Resume on stopped WorkQueue. Use IsSuspended to check state.");
                case GateOpened:
                    throw new InvalidOperationException("Cannot call Resume on WorkQueue unless its workers have been suspended. Use IsSuspended to check state.");
                default:
                    throw new InvalidOperationException("Invalid WorkQueue state detected.");
            }
            // if any WorkQueue.Resumed has any listeners notify them that workers have been resumed
            if (this.Resumed != null)
            {
                new Task(() => { this.Resumed(this); })
                    .Start();
            }
            return true;
        }
        /// <summary>
        /// Signals all workers to stop. Non-blocking
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public void Stop()
        {
            this.ValidateCallingThread();
            // close the worker gate and signal all workers that their shift is over
            Interlocked.Exchange(ref _workerGate, GateClosed);
            lock (_workQueue)
            {
                Monitor.PulseAll(_workQueue);
            }
            // asynchronously notify all WorkQueue.Stopped listeners that the queue has stopped
            new Task(() =>
                {
                    if (this.Stopped != null)
                    {
                        // wait for all workers to stop
                        lock (_workQueue)
                        {
                            while (_workingWorkers > 0)
                            {
                                Monitor.Wait(_workQueue);
                            }
                        }
                        // call the listeners
                        this.Stopped(this);
                    }
                })
                .Start();
        }
        /// <summary>
        /// Signals all workers to stop and blocks the calling thread until all work has stopped
        /// </summary>
        /// <param name="millisecondsTimeout">The maximum length of time (in milliseconds) the caller is willing to wait</param>
        /// <returns>Returns true if all work has stopped; false if the wait timed out and work might still be happening</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Stop(int millisecondsTimeout)
        {
            this.ValidateCallingThread();
            // signal all workers to stop, then wait for them to complete
            this.Stop();
            return this.Wait(millisecondsTimeout);
        }
        /// <summary>
        /// Signals all workers to suspend. Non-blocking
        /// </summary>
        /// <returns>Returns true if all work has been suspended; false if the wait timed out and work might still be happening</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Suspend()
        {
            this.ValidateCallingThread();
            //
            switch (Interlocked.CompareExchange(ref _workerGate, GateBlocked, GateOpened))
            {
                case GateBlocked:
                    return true;
                case GateClosed:
                    throw new InvalidOperationException("Cannot suspend a stopped WorkQueue. Use IsRunning to check state.");
                case GateOpened:
                    {
                        new Task(() =>
                        {
                            // if any WorkQueue.Suspended has any listeners notify them that workers have been resumed once the queue has been suspended
                            if (this.Suspended != null)
                            {
                                lock (_workQueue)
                                {
                                    while (Interlocked.Read(ref _workingWorkers) > 0)
                                    {
                                        Monitor.Wait(_workQueue);
                                    }
                                }
                                this.Suspended(this);
                            }
                        })
                        .Start();
                        return true;
                    }
                default:
                    throw new InvalidOperationException();
            }
        }
        /// <summary>
        /// Blocks the calling thread until all work has been suspended or the timeout has expired
        /// </summary>
        /// <param name="millisecondsTimeout">The maximum length of time (in milliseconds) the caller is willing to wait</param>
        /// <returns>Returns true if all work has been suspended; false if the wait timed out and work might still be happening</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Suspend(int millisecondsTimeout)
        {
            this.ValidateCallingThread();
            //
            switch (Interlocked.CompareExchange(ref _workerGate, GateBlocked, GateOpened))
            {
                case GateBlocked:
                    return true;
                case GateClosed:
                    throw new InvalidOperationException("Cannot suspend a stopped WorkQueue. Use IsRunning to check state.");
                case GateOpened:
                    {
                        Stopwatch timer = new Stopwatch();
                        lock (_workQueue)
                        {
                            Monitor.PulseAll(_workQueue);
                            //
                            timer.Start();
                            while (Interlocked.Read(ref _workingWorkers) > 0)
                            {
                                if (millisecondsTimeout == Timeout.Infinite)
                                {
                                    Monitor.Wait(_workQueue, Timeout.Infinite);
                                }
                                else
                                {
                                    if (timer.ElapsedMilliseconds >= millisecondsTimeout || !Monitor.Wait(_workQueue, (int)(millisecondsTimeout - timer.ElapsedMilliseconds)))
                                        return false;
                                }
                            }
                            timer.Stop();
                        }
                    } break;
                default:
                    throw new InvalidOperationException();
            }
            // if any WorkQueue.Suspended has any listeners notify them that workers have been suspended
            if (this.Suspended != null)
            {
                new Task(() => { this.Suspended(this); })
                    .Start();
            }
            return true;
        }
        /// <summary>
        /// Blocks the calling thread until all work has stopped or been suspended
        /// </summary>
        /// <param name="millisecondsTimeout">The maximum length of time (in milliseconds) the caller is willing to wait</param>
        /// <returns>Returns true if all work has stopped; false if the wait timed out and work might still be happening</returns>
        /// <exception cref="System.InvalidOperationException">
        /// Throws an InvalidOperationException if accessed from a thread other than the thread it was created on
        /// </exception>
        public bool Wait(int millisecondsTimeout)
        {
            this.ValidateCallingThread();
            //
            Stopwatch timer = new Stopwatch();
            lock (_workQueue)
            {
                timer.Start();
                while ((Interlocked.Read(ref _workerGate) != GateClosed && _workQueue.Count > 0)
                        || Interlocked.Read(ref _workingWorkers) > 0)
                {
                    if (millisecondsTimeout == Timeout.Infinite)
                    {
                        Monitor.Wait(_workQueue, Timeout.Infinite);
                    }
                    else
                    {
                        if (timer.ElapsedMilliseconds >= millisecondsTimeout || !Monitor.Wait(_workQueue, (int)(millisecondsTimeout - timer.ElapsedMilliseconds)))
                            return false;
                    }
                }
                timer.Stop();
            }
            return true;
        }
        #region Private Methods
        private void ValidateCallingThread()
        {
            if (Thread.CurrentThread.ManagedThreadId != _masterThreadId)
                throw new InvalidOperationException("Cross-thread operation not valid: WorkQueue accessed from a thread other than the thread it was created on.");
        }

        private void ValidateWorkerThread()
        {
            lock (_workerThreads)
            {
                if (!_workerThreads.ContainsKey(Thread.CurrentThread.ManagedThreadId))
                    throw new InvalidOperationException("Cross-thread operation not valid: WorkQueue.Check accessed from a thread other than a member worker thread.");
            }
        }

        private void Work()
        {
            TWork work = default(TWork);
            //
            while (true)
            {
                lock (_workerThreads)
                {
                    if (_workerThreads.Count > _maxWorkers)
                        break;
                }
                lock (_workQueue)
                {
                    // suspended loop keeps threads trapped here so long as the queue is suspended
                    while (Interlocked.Read(ref _workerGate) == GateBlocked)
                    {
                        Monitor.PulseAll(_workQueue);
                        Monitor.Wait(_workQueue);
                    }
                    lock (_workerThreads)
                    {
                        if (_workFetchMethod != null && _workQueue.Count <= _minCapacity)
                        {
                            if (Interlocked.Exchange(ref _fetchGate, GateClosed) == GateOpened)
                            {
                                new Task(() =>
                                {
                                    foreach (TWork newItem in _workFetchMethod(_maxCapacity - _workQueue.Count))
                                    {
                                        this.AddWork(newItem);
                                    }
                                    Interlocked.Exchange(ref _fetchGate, GateOpened);
                                }).Start();
                            }
                        }
                    }
                    while (_workQueue.Count == 0 && Interlocked.Read(ref _workerGate) != GateClosed)
                    {
                        Monitor.Wait(_workQueue, Timeout.Infinite);
                    }
                    if (Interlocked.Read(ref _workerGate) == GateClosed)
                    {
                        Monitor.PulseAll(_workQueue);
                        break;
                    }
                    work = _workQueue.Dequeue();
                }
                // enter the work block, keeping track of the number of working workers
                Interlocked.Increment(ref _workingWorkers);
                {
                    try
                    {
                        _workerMethod(work);
                    }
                    catch (Exception exception)
                    {
                        lock (_workerThreads)
                        {
                            if (_exceptionHandler == null)
                            {
                                this.Stop();
                                throw exception;
                            }
                            // let the external exception handler, handle the exception
                            new Task(() => { this._exceptionHandler(exception, work); }).Start();
                        }
                    }
                }
                Interlocked.Decrement(ref _workingWorkers);
                // pulse the queue to let it know work has been completed
                lock (_workQueue)
                {
                    Monitor.PulseAll(_workQueue);
                }
            }
            // when a worker thread leaves the main loop, it should be removed from the pool
            lock (_workerThreads)
            {
                _workerThreads.Remove(Thread.CurrentThread.ManagedThreadId);
            }
        }
        #endregion
        #endregion
    }
}
